import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class Demo03_window {

    /**
     * 非按key分区窗口
     * 就是所有的数据都进入到同一个分区中
     */
    public static void nonKeyedWindow(DataStream<Demo02_sensorReading> sensorDataStream){

        //窗口的标准形式
        AllWindowedStream<Demo02_sensorReading, TimeWindow> tumblingWindow
                = sensorDataStream.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        //时间窗口的简化形式已经弃用
        AllWindowedStream<Demo02_sensorReading, TimeWindow> tumblingWindowEasy
                = sensorDataStream.timeWindowAll(Time.seconds(5));
        AllWindowedStream<Demo02_sensorReading, TimeWindow> slidingWindowEasy
                = sensorDataStream.timeWindowAll(Time.seconds(10),Time.seconds(15));

        //计数窗口的简化形式还可以继续使用
        AllWindowedStream<Demo02_sensorReading, GlobalWindow> tumblingCountEasy
                = sensorDataStream.countWindowAll(5);
        AllWindowedStream<Demo02_sensorReading, GlobalWindow> slidingCountEasy
                  = sensorDataStream.countWindowAll(8,5);

        slidingCountEasy.reduce(
                (value1, value2)-> value2
        ).print();
    }

    /**
     * 按key分区窗口
     * @param sensorDataStream
     */
    public static void keyedWindow(DataStream<Demo02_sensorReading> sensorDataStream){
        //滚动时间窗口
//        WindowedStream<Demo02_sensorReading, String, TimeWindow> tumblingTimeWindow = sensorDataStream
//                .keyBy(Demo02_sensorReading::getSensorId)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        //滑动时间窗口
//        WindowedStream<Demo02_sensorReading, String, TimeWindow> slidingTimeWindow = sensorDataStream
//                .keyBy(Demo02_sensorReading::getSensorId)
//                .window(SlidingProcessingTimeWindows.of(Time.seconds(8),Time.seconds(5)));

        //会话时间窗口
//        WindowedStream<Demo02_sensorReading, String, TimeWindow> sessionTimeWindow = sensorDataStream
//                .keyBy(Demo02_sensorReading::getSensorId)
//                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

        //滚动计数窗口，在全局窗口的基础上，添加触发器，含义是到多少条数据的时候触发窗口关闭
//        WindowedStream<Demo02_sensorReading, String, GlobalWindow> tumblingCountWindow = sensorDataStream
//                .keyBy(Demo02_sensorReading::getSensorId)
//                .window(GlobalWindows.create())
//                .trigger(CountTrigger.of(5));

        //滚动计数窗口，在全局窗口的基础上，添加触发器，含义是到多少条数据的时候触发窗口关闭
//        WindowedStream<Demo02_sensorReading, String, GlobalWindow> slidlingCountWindow = sensorDataStream
//                .keyBy(Demo02_sensorReading::getSensorId)
//                .window(GlobalWindows.create())
//                .evictor(CountEvictor.of(8))
//                .trigger(CountTrigger.of(5));

        //滚动计数窗口的简写形式
        WindowedStream<Demo02_sensorReading, String, GlobalWindow> tumblingCountWindowEasy = sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .countWindow(5);

        //滑动计数窗口的简写形式
        WindowedStream<Demo02_sensorReading, String, GlobalWindow> slidingCountWindowEasy = sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .countWindow(8,5);

        slidingCountWindowEasy.reduce(
                (value1, value2)-> value2
        ).print();

    }

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.setParallelism(1);

        //接收String类型的原始数据
        DataStreamSource<String> dataStream = Demo02_source.source_socket(env,9999);
        //把String类型转成SensorReading类型
        DataStream<Demo02_sensorReading> sensorDataStream = Demo02_transformation.transformation_simple(dataStream);

        keyedWindow(sensorDataStream);

        try {
            env.execute("window");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
